import java.io.IOException;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class Step2 {

	/** Input: (<obj>, <"id	url|zeta|k|node_1|..|node_n">)
	 *  where obj: 	     not significant
	 * 		  id: 	     id of a node (it's an hash), let's call it j
	 * 		  url: 	     url (or string identified) of j
	 * 		  zeta:	     # of received visits to j
	 * 		  k:	     # of live coupons (in this phase should be 0)
	 * 		  node_1..n: adjacency list 
	 * 	
	 *  Output: (<url>, <pagerank>) 
	 *  
	 *  Note: The InputFormat KeyValueTextInputFormat isn't no more
	 *  available: the TextInputFormat must be used doing the split manually. 
	 */
	//this last mapper is in charge to comput the final value of pagerank for each graph node
	public static class step2_mapper extends Mapper<Object, Text, Text, Text> {
		private Text outputkey = new Text();
		private Text outputvalue = new Text();

		public void map(Object key, Text value, Context context)
				throws IOException, InterruptedException {
			String[] input_value = value.toString().split("[|]");
			System.out.println("Value: " + value.toString());
			String url=(input_value[0].split("\t"))[1];
			//Eacy field is read by the input tuples
			float zeta= Float.parseFloat(input_value[1]);
			float epsilon=context.getConfiguration().getFloat("epsilon", 0);
			long cnlogn=context.getConfiguration().getLong("cnlogn", 0);
			outputkey.set(url);
			
			//This is the real computation of the pagerank according to the algorithm
			System.out.println("zeta: " + zeta + "; epsilon: " + epsilon + "; cnlogn: " + cnlogn);
			float pagerank=(zeta*epsilon)/(cnlogn);
			outputvalue.set(""+pagerank);
			context.write(outputkey, outputvalue);
		}
	}
}
